package com.bw.medical.realtime.app.dwd;

import com.bw.medical.realtime.common.MedicalCommon;
import com.bw.medical.realtime.util.CreateEnvUtil;
import com.bw.medical.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
9.8 互动域用户评价事务事实表
 */

public class DwdInteractionReview {
    public static void main(String[] args) {
        // TODO 1. 初始化表处理环境
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(8089, "medical_dwd_interaction_review");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        // TODO 2. 读取数据
        String topic = MedicalCommon.KAFKA_ODS_TOPIC;
        String groupId = "medical_dwd_interaction_review";
        CreateEnvUtil.createOdsTable(tableEnv, topic, groupId);

        // TODO 3. 筛选数据
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "`data`['id'] `id`,\n" +
                "`data`['update_time'] `review_time`,\n" +
                "`data`['rating'] `rating`,\n" +
                "`data`['doctor_id'] `doctor_id`,\n" +
                "`data`['patient_id'] `patient_id`,\n" +
                "`data`['user_id'] `user_id`\n" +
                "from `topic_db`\n" +
                "where `table` = 'consultation'\n" +
                "and `type` = 'update'\n" +
                "and `data`['status'] = '207'");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 4. 写入 Kafka 目标主题
        String sinkTopic = "medical_dwd_interaction_review";
        tableEnv.executeSql("create table " + sinkTopic + "(\n" +
                "`id` string,\n" +
                "`review_time` string,\n" +
                "`rating` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string,\n" +
                "`user_id` string" +
                ") " + KafkaUtil.getKafkaSinkDDL(sinkTopic));
        tableEnv.executeSql("insert into " + sinkTopic + " select * from result_table");
    }
}
